package org.tao.lightningmq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.dto.ServiceRegistryRespDTO;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.enums.ReplicationMsgTypeEnum;
import org.tao.lightningmq.nameserver.event.model.RegistryEvent;
import org.tao.lightningmq.nameserver.event.model.ReplicationMsgEvent;
import org.tao.lightningmq.nameserver.store.ServiceInstance;

/**
 * @Author lht
 * @date 2025/1/7 - 19:58
 * @description: 注册事件监听器
 */
public class RegistryListener implements Listener<RegistryEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RegistryListener.class);

    @Override
    public void onReceive(RegistryEvent event) throws Exception {
        ChannelHandlerContext context = event.getContext();
        ServiceRegistryRespDTO registryRespDTO = new ServiceRegistryRespDTO();
        registryRespDTO.setMsgId(event.getMsgId());
        // 进行安全认证
        if (!isVerify(event.getUser(), event.getPassword())) {
            TcpMsg tcpMsg = new TcpMsg(NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode(),
                    JSON.toJSONBytes(registryRespDTO));
            context.writeAndFlush(tcpMsg);
            context.close();
            throw new IllegalAccessException("error account to connected!");
        }

        context.attr(AttributeKey.valueOf("reqId")).set(event.getIp() + ":" + event.getPort());
        ServiceInstance instance = new ServiceInstance(event);
        CommonCache.getServiceInstanceManager().put(instance);

        // 判断是否是单机架构
        if (CommonCache.getReplicationMode() == null) {
            TcpMsg tcpMsg = new TcpMsg(NameServerResponseCode.REGISTRY_SUCCESS,
                    JSON.toJSONBytes(registryRespDTO));
            context.writeAndFlush(tcpMsg);
            LOGGER.info("单机模式下实例注册成功:{}", instance);
            return;
        }

        // 说明是主从模式，需要同步给从节点
        ReplicationMsgEvent replicationMsgEvent = new ReplicationMsgEvent();
        replicationMsgEvent.setServiceInstance(instance);
        // 這個地方的msgId直接使用注册事件的msgId，用于后续收到所有从节点的ACK后返回给对于的broker进行确认
        replicationMsgEvent.setMsgId(event.getMsgId());
        replicationMsgEvent.setContext(event.getContext());
        replicationMsgEvent.setType(ReplicationMsgTypeEnum.REGISTRY.getCode());
        CommonCache.getReplicationMsgQueueManager().put(replicationMsgEvent);
    }

    public static boolean isVerify(String user, String password) {
        String rightUser = CommonCache.getNameserverProperties().getNameserverUser();
        String rightPassword = CommonCache.getNameserverProperties().getNameserverPwd();
        return rightUser.equals(user) && rightPassword.equals(password);
    }

}
